package com.atguigu.gmall.realtime.dim.function;/**
 * @Author: Wxs
 * @Description: TODO
 * @Date: 2024/12/14 10:40
 */

import com.alibaba.fastjson.JSONObject;
import com.group2.edu.realtime.common.bean.TableProcessDim;
import com.group2.edu.realtime.common.constant.Constant;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.util.*;

/**
 * @Author: Wxs
 * @Description: TODO
 * @Date: 2024/12/14 10:40
 */
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>> {
    private Map<String,TableProcessDim> configMap = new HashMap<>();
    private MapStateDescriptor<String, TableProcessDim> mapStateDescriptor;

    public TableProcessFunction(MapStateDescriptor<String, TableProcessDim> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //将配置表中的配置信息预加载到configMap
        //注册驱动
        Class.forName("com.mysql.cj.jdbc.Driver");
        //建立连接
        java.sql.Connection conn = DriverManager.getConnection(Constant.MYSQL_URL, Constant.MYSQL_USER_NAME, Constant.MYSQL_PASSWORD);
        //获取数据库操作对象
        String sql = "select * from gmall_config.table_process_dim";
        PreparedStatement ps = conn.prepareStatement(sql);
        //执行SQL语句
        ResultSet rs = ps.executeQuery();
        ResultSetMetaData metaData = rs.getMetaData();
        //处理结果集
        while (rs.next()){
            //定义一个json对象，用于接收查询出来的一条结果
            JSONObject jsonObj = new JSONObject();
            for (int i = 1; i <= metaData.getColumnCount(); i++) {
                String columnName = metaData.getColumnName(i);
                Object columnValue = rs.getObject(i);
                jsonObj.put(columnName,columnValue);
            }
            //将jsonObj转换为实体类对象
            TableProcessDim tableProcessDim = jsonObj.toJavaObject(TableProcessDim.class);
            configMap.put(tableProcessDim.getSourceTable(),tableProcessDim);
        }

        //释放资源
        rs.close();
        ps.close();
        conn.close();

    }

    //处理主流数据
    @Override
    public void processElement(JSONObject jsonObject, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.ReadOnlyContext readOnlyContext, Collector<Tuple2<JSONObject, TableProcessDim>> collector) throws Exception {
        //{"database":"gmall0620","xid":13275,"data":{"tm_name":"Redmi","create_time":"2021-12-14 00:00:00","logo_url":"aaa","id":1},"old":{},"commit":true,"type":"update","table":"base_trademark","ts":1733045365}
        //System.out.println(jsonObject);
        ReadOnlyBroadcastState<String, TableProcessDim> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);
        String table = jsonObject.getString("table");
        TableProcessDim tableProcessDim =null;
        if ((tableProcessDim = broadcastState.get(table)) != null
                || (tableProcessDim = configMap.get(table))!= null){
            JSONObject data = jsonObject.getJSONObject("data");
            //在向下游传递数据之前，过滤掉不需要的数据
            String sinkColumns = tableProcessDim.getSinkColumns();
            deleteNotNeedColumns(data,sinkColumns);
            //补充type
            String type = jsonObject.getString("type");
            data.put("type",type);
            collector.collect(Tuple2.of(data,tableProcessDim));
        }


    }

    //处理广播流配置信息
    @Override
    public void processBroadcastElement(TableProcessDim tp, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.Context context, Collector<Tuple2<JSONObject, TableProcessDim>> collector) throws Exception {
        //获取操作信息
        String op = tp.getOp();
        //获取广播状态
        BroadcastState<String, TableProcessDim> broadcastState = context.getBroadcastState(mapStateDescriptor);
        //获取表名（广播流中的k）
        String sourceTable = tp.getSourceTable();
        if ("d".equals(op)){
            broadcastState.remove(sourceTable);
            configMap.remove(sourceTable);
        }else {
            broadcastState.put(sourceTable,tp);
            configMap.put(sourceTable,tp);
        }

    }
    private void deleteNotNeedColumns(JSONObject data, String sinkColumns) {
        //sinkColumns:id,tm_name
        List<String> columnList = Arrays.asList(sinkColumns.split(","));
        Set<Map.Entry<String, Object>> entries = data.entrySet();
        entries.removeIf(entry -> !columnList.contains(entry.getKey()));


    }

}
